package com.epoint.flinkdemo.transform;

import com.epoint.flinkdemo.entity.HcpEvainstance;
import com.epoint.flinkdemo.utils.DataSourceUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * @author liufl
 * @version [版本号, 21-4-29]
 */
public class AllProcessFunction extends ProcessFunction<HcpEvainstance, HcpEvainstance>
{

    private final static String TASKR_SQL = "select wditemName, processcode, processname, taskstatus,sxstatus from hcp_process_standard_task where wditemName = ?";

    private static final String FWZN_SQL = "select item_code, catalog_id, catalog_code, ql_name from audit_fwzn where item_code=?";

    private static final String QLCODE_SQL = "select qlcode, ql_name, areacode from audit_ql where qlcode like ?";

    private static final String LOGCATALOGID_SQL = "select rowguid, catalog_id, parentguid,ql_code,ql_name from audit_catalog where catalog_id = ?";

    @Override
    public void processElement(HcpEvainstance value, Context context, Collector<HcpEvainstance> collector)
            throws Exception {
        // wditemname == null
        if (StringUtils.isNotBlank(value.getWditemname())) {
            if (taskrWditemnameIsEmpty(value)) {
                collector.collect(value);
            }
            else {
                handle1(value, collector);
            }
        }
        else {
            handle1(value, collector);
        }
    }

    /**
     * wditemId是否为空
     *
     * @param value
     * @return
     */
    public boolean wditemIdIsEmpty(HcpEvainstance value) {
        return StringUtils.isBlank(value.getWditemid());
    }

    /**
     * taskr.wditemName是否为空
     *
     * @param wditemName
     * @return
     */
    public boolean taskrWditemnameIsEmpty(HcpEvainstance value) throws SQLException {
        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            connection = DataSourceUtils.getDataSource().getConnection();
            ps = connection.prepareStatement(TASKR_SQL);

            ps.setString(1, value.getWditemname());
            rs = ps.executeQuery();
            if (rs.next()) {
                String wditemName = rs.getString("wditemName");
                if (StringUtils.isNotBlank(wditemName)) {
                    value.setTaskcode(rs.getString("processcode"));
                    value.setTaskname(rs.getString("processcodename"));
                    value.setTaskstatus(rs.getInt("taskstatus"));
                    value.setSxstatus(rs.getInt("sxstatus"));
                    return true;
                }
            }

        }
        finally {
            DataSourceUtils.closeDataSource(connection, ps, rs);
        }
        return false;
    }

    /**
     * fwzn.item_code == eval.wditemId
     *
     * @param value
     * @return
     */
    public String fwznItemCodeIsEmpty(HcpEvainstance value) throws Exception {
        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            connection = DataSourceUtils.getDataSource().getConnection();
            ps = connection.prepareStatement(FWZN_SQL);

            ps.setString(1, value.getWditemid());
            rs = ps.executeQuery();
            if (rs.next()) {
                String itemCode = rs.getString("item_code");
                if (StringUtils.isNotBlank(itemCode)) {
                    String catalogcode = rs.getString("catalog_code");
                    if (catalogcode.endsWith("000")) {
                        value.setTaskcode(catalogcode.substring(0, 9));
                        value.setTaskname(rs.getString("ql_name"));
                        value.setSxstatus(1);
                        value.setTaskstatus(2);
                        return "1";
                    }
                    else {
                        String catalogid = rs.getString("catalog_id");
                        value.setFwznCatalogCode(catalogcode);
                        value.setFwznCatalogId(catalogid);
                        if (StringUtils.isNotBlank(catalogid)) {
                            return "2";
                        }
                        else {
                            return "3";
                        }
                    }
                }
            }
            return "0";
        }
        finally {
            DataSourceUtils.closeDataSource(connection, ps, rs);
        }
    }

    public void handle1(HcpEvainstance value, Collector<HcpEvainstance> collector) throws Exception{
        if (wditemIdIsEmpty(value)) {
            value.setSxstatus(0);
            collector.collect(value);
        }
        else {
            String fwznItemCodeResult = fwznItemCodeIsEmpty(value);
            if ("0".equals(fwznItemCodeResult)) {
                value.setSxstatus(0);
                collector.collect(value);
            }
            // 以"000"结尾
            else if ("1".equals(fwznItemCodeResult)) {
                collector.collect(value);
            }
            // 不以"000"结尾 且 catalogid is null
            else if ("2".equals(fwznItemCodeResult)) {
                handleQlCode(value);
                collector.collect(value);
            }
            // 不以"000"结尾 且 catalogid is not null
            else if ("3".equals(fwznItemCodeResult)) {
                handleCatalogID(value);
                collector.collect(value);
            }
        }
    }

    public String handleQlCode(HcpEvainstance value) throws Exception {
        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            connection = DataSourceUtils.getDataSource().getConnection();
            ps = connection.prepareStatement(QLCODE_SQL);

            ps.setString(1, value.getFwznCatalogCode().substring(0, 9));
            rs = ps.executeQuery();
            if (rs.next()) {
                String qlcode = rs.getString("qlcode");
                String areacode = rs.getString("areacode");
                String qlname = rs.getString("ql_name");
                if (StringUtils.isNotBlank(qlcode)) {
                    if ("310000".equals(areacode) || "310100".equals(areacode)) {
                        value.setTaskcode(qlcode.substring(0,9));
                        value.setTaskname(qlname);
                        value.setSxstatus(1);
                        value.setTaskstatus(2);
                        return "1";
                    }
                    else if (!"310000".equals(areacode) || !"310100".equals(areacode)) {
                        if ("999999".equals(areacode)) {
                            value.setTaskcode(qlcode.substring(0,9));
                            value.setTaskname(qlname);
                            value.setSxstatus(1);
                            value.setTaskstatus(2);
                            return "2";
                        }
                        else {
                            value.setSxstatus(0);
                            return "0";
                        }
                    }
                }
                else {
                    value.setSxstatus(0);
                    return "0";
                }
            }
            value.setSxstatus(0);
            return "0";
        } finally {
            DataSourceUtils.closeDataSource(connection, ps, rs);
        }
    }

    public String handleCatalogID(HcpEvainstance value) throws Exception {
        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            connection = DataSourceUtils.getDataSource().getConnection();
            ps = connection.prepareStatement(LOGCATALOGID_SQL);

            ps.setString(1, value.getFwznCatalogId());
            rs = ps.executeQuery();
            if (rs.next()) {
                String catalogid = rs.getString("catalog_id");
                String parentguid = rs.getString("areacode");
                String qlname = rs.getString("ql_name");
                String qlcode = rs.getString("qlcode");
                String rowguid = rs.getString("rowguid");
                if (StringUtils.isNotBlank(catalogid)) {
                    if ("self".equals(parentguid)) {
                        value.setTaskcode(qlcode.substring(0, 9));
                        value.setTaskname(qlname);
                        value.setSxstatus(1);
                        value.setTaskstatus(2);
                        return "1";
                    }
                }
                else {
                    if (rowguid.equals(parentguid)) {
                        value.setTaskcode(qlcode.substring(0, 9));
                        value.setTaskname(qlname);
                        value.setSxstatus(1);
                        value.setTaskstatus(2);
                        return "2";
                    }
                    else {
                        value.setSxstatus(0);
                        return "0";
                    }
                }
            }
            value.setSxstatus(0);
            return "0";
        } finally {
            DataSourceUtils.closeDataSource(connection, ps, rs);
        }
    }
}
